package net.bw.realtime.warehouse.dwd;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @BelongsProject: realtime-project-10zlq
 * @BelongsPackage: net.bw.realtime.warehouse.dwd
 * @Author: zhangleqing
 * @CreateTime: 2025-08-26  11:22
 * @Description: TODO dwd_summary_cars_log (车辆进出汇总表)
 * @Version: 1.0
 */
public class DwdCarsLog {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 设置检查点  严重影响代码运行的速度，但是自动容错还是挺不错的
//        env.enableCheckpointing(5000);
//        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        // todo 1.车辆进出事实表 ods_cars_log_clean
        tEnv.executeSql(
                "CREATE TABLE ods_cars_log_clean (\n" +
                        " id STRING,\n" +
                        " opTime TIMESTAMP(3),\n" +
                        " ctype STRING,\n" +
                        " carCode STRING,\n" +
                        " cId BIGINT,\n" +
                        " proc_time AS PROCTIME(),\n" +
                        " WATERMARK FOR opTime AS opTime - INTERVAL '5' SECOND\n" +
                        ") WITH (\n" +
                        " 'connector' = 'kafka',\n" +
                        " 'topic' = 'ods_cars_log_clean',\n" +
                        " 'properties.bootstrap.servers' = 'node101:9092',\n" +
                        " 'properties.group.id' = 'ods_cars_log_clean',\n" +
                        " 'scan.startup.mode' = 'earliest-offset',\n" +
                        " 'format' = 'json'\n" +
                        ")"
        );
        // 测试水位线是否可用
/*
SELECT
    carCode,
    COUNT(*) AS car_count,
    TUMBLE_START(opTime, INTERVAL '1' HOUR) AS window_start,
    TUMBLE_END(opTime, INTERVAL '1' HOUR) AS window_end
FROM ods_cars_log_clean
GROUP BY
carCode, TUMBLE(opTime, INTERVAL '1' HOUR) order by car_count limit 10
 */


/*
CREATE TABLE ods_cars_clean(
    id INT,
    owerId INT,
    carCode STRING,
    carColor STRING,
    type STRING,
    remark STRING
) WITH (
    'connector' = 'kafka',
    'topic' = 'ods_cars_clean',
    'properties.bootstrap.servers' = 'node101:9092',
    'properties.group.id' = 'ods_cars_clean',
    'scan.startup.mode' = 'earliest-offset',
    'format' = 'json'
)
 */
        // todo 2.搭建车辆维度表 ods_cars_clean
        tEnv.executeSql(
                "CREATE TABLE ods_cars_clean(\n" +
                        "    id INT,\n" +
                        "    owerId INT,\n" +
                        "    carCode STRING,\n" +
                        "    carColor STRING,\n" +
                        "    type STRING,\n" +
                        "    remark STRING\n" +
                        ") WITH ( \n" +
                        "    'connector' = 'kafka',\n" +
                        "    'topic' = 'ods_cars_clean',\n" +
                        "    'properties.bootstrap.servers' = 'node101:9092',\n" +
                        "    'properties.group.id' = 'ods_cars_clean',\n" +
                        "    'scan.startup.mode' = 'earliest-offset',\n" +
                        "    'format' = 'json'\n" +
                        ")"
        );

/*
  TUMBLE_START(opTime, INTERVAL '5' SECOND) AS window_start,
  TUMBLE_END(opTime, INTERVAL '5' SECOND) AS window_end

SELECT
  w.id,
  COALESCE(d.owerId, -1) AS owerId,
  w.opTime,
  w.cId,
  w.carCode,
  w.ctype,
  w.window_start,
  w.window_end
FROM
  TABLE(TUMBLE(TABLE ods_cars_log_clean, DESCRIPTOR(opTime), INTERVAL '5' SECOND)) AS w
LEFT JOIN ods_cars_clean AS d
ON w.cId = d.id
 */
        // todo 3.创建 dwd_summary_cars_log 表  车辆进出汇总表
        // 使用窗口TVF方案（Flink 1.13+）
        Table dwd_summary_cars_log = tEnv.sqlQuery(
                "SELECT\n" +
                        "  w.id,\n" +  // 使用 w 作为窗口表的别名
                        "  COALESCE(d.owerId, -1) AS owerId,\n" +
                        "  w.opTime,\n" +
                        "  w.cId,\n" +
                        "  w.carCode,\n" +
                        "  w.ctype,\n" +
                        "  w.window_start,\n" +  // 直接使用窗口列
                        "  w.window_end\n" +      // 直接使用窗口列
                        "FROM \n" +
                        "  TABLE(TUMBLE(TABLE ods_cars_log_clean, DESCRIPTOR(opTime), INTERVAL '5' SECOND)) AS w\n" +
                        "LEFT JOIN ods_cars_clean AS d\n" +
                        "ON w.cId = d.id"
        );

        tEnv.createTemporaryView("dwd_summary_cars_log", dwd_summary_cars_log);


        tEnv.sqlQuery("select * from dwd_summary_cars_log limit 10")
                .execute().print();



        env.execute("DwdCarsLog");
    }
}
